package com.lanzuo.pana.adapter.client;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.lanzuo.adapter.common.constant.CommandConstant;
import com.lanzuo.pana.adapter.client.decoder.JsonMessageDecoder;
import com.lanzuo.pana.adapter.client.encoder.JsonMessageEncoder;
import com.lanzuo.pana.adapter.client.handler.MessageHandler;
import com.lanzuo.pana.adapter.client.watcher.ChangeoverWatcher;
import com.lanzuo.pana.adapter.client.watcher.SshConfig;
import com.lanzuo.pana.adapter.config.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author <a href="eric_zheng@lanzuo.com.cn">eric</a>
 * @date 2025/08/12 13:18
 * @description
 */
@Slf4j
public class PanaMounterMinaClient {
    public static final AtomicReference<IoSession> ACTIVE_SESSION = new AtomicReference<>();
    // 3秒重试
    private static final long RECONNECT_DELAY_SECONDS = 3;

    private final NioSocketConnector connector;
    private final AtomicBoolean isStopped = new AtomicBoolean(false);


    /**
     * 初始化客户端配置
     *
     * @return 客户端
     */
    public PanaMounterMinaClient() {
        NioSocketConnector connector = new NioSocketConnector();
        // 添加编解码器
        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new JsonMessageEncoder(), new JsonMessageDecoder()));

        // 注册过滤器
        connector.getFilterChain().addLast("registration-filter", new RegistrationFilter());

        connector.setHandler(new MessageHandler());
        connector.getSessionConfig().setReadBufferSize(2048);
        connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 20);

        this.connector = connector;
    }

    public static void main(String[] args) {
        PanaMounterMinaClient client = new PanaMounterMinaClient();
        client.start();
    }

    /**
     * 创建客户端即连接，随即发送注册指令
     */
    public void start() {
        String eapIp = SpringContextHolder.getProperty("eap.ip", "");
        int eapPort = SpringContextHolder.getProperty("eap.port", Integer.class, null);
        while (!isStopped.get()) {
            try {
                ConnectFuture future = connector.connect(new InetSocketAddress(eapIp, eapPort));
                future.awaitUninterruptibly();

                if (future.isConnected()) {
                    log.info("✅ 连接成功: {}", future.getSession().getRemoteAddress());

                    // 等待会话关闭（说明断开了）
                    future.getSession().getCloseFuture().awaitUninterruptibly();
                    log.warn("⚠️ 连接已断开，准备重连...");
                } else {
                    log.error("❌ 连接失败: 无法连接到 {}:{}", eapIp, eapPort);
                }
            } catch (Exception e) {
                log.error("💥 连接异常: {}", e.getMessage());
            }

            // 检查是否已停止
            if (isStopped.get()) break;
            // 等待一段时间后重试
            try {
                log.info("⏳ {} 秒后重试...", RECONNECT_DELAY_SECONDS);
                TimeUnit.SECONDS.sleep(RECONNECT_DELAY_SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }

        log.warn("🛑 客户端已停止");
    }

    /**
     * 停止客户端
     */
    public void stop() {
        isStopped.set(true);
        connector.dispose();
    }

    /**
     * @author eric
     */
    @Slf4j
    public static class RegistrationFilter extends IoFilterAdapter {
        // 创建只有一个线程的线程池
        ExecutorService singleExecutor = Executors.newSingleThreadExecutor();

        @Override
        public void sessionOpened(NextFilter nextFilter, IoSession session) {
            ACTIVE_SESSION.set(session);
            log.info("📝 [RegistrationFilter] 发送注册指令...");

            String ip = SpringContextHolder.getProperty("lnb.pana.ip", "");
            String nuiqueCode = SpringContextHolder.getProperty("lnb.pana.nuique_code", "");

            String registerJson = StrUtil.format(CommandConstant.REGISTER_TEMPLATE, ip, nuiqueCode, RandomUtil.randomString(10));
            // 会经过 codec 自动加 \0
            session.write(registerJson);
            nextFilter.sessionOpened(session);
        }

        @Override
        public void messageReceived(NextFilter nextFilter, IoSession session, Object message) {
            String msg = (String) message;
            if (msg.contains("ReplyRegist")) {
                log.info("✅ [RegistrationFilter] 收到注册响应: {}", msg);

                JSONObject replyRegist = JSONUtil.parseObj(msg);
                if (replyRegist.containsKey("Data")) {
                    JSONObject replyRegistData = replyRegist.getJSONObject("Data");

                    JSONObject machineInfo = replyRegistData.getJSONObject("MachineInfo");
                    String machineCode = machineInfo.getStr("machineCode");
                    session.setAttribute("registered", true);

                    if (machineCode.contains("#")) {
                        String[] machineSplit = machineCode.split("#");
                        // 可解析 token、userId 等，存入 session
                        session.setAttribute("machineCode", machineSplit[0]);
                        if (machineSplit.length == 2) {
                            session.setAttribute("machinePosition", machineSplit[1]);
                        }

                        // 开启监听程序变化
                        singleExecutor.submit(new ChangeoverWatcher(SshConfig.newConfig(), session));
                    }
                }

                // 继续传递给后续 filter（如 heartbeat）
                nextFilter.messageReceived(session, message);
            } else {
                // 非注册响应，直接传递
                nextFilter.messageReceived(session, message);
            }
        }

        @Override
        public void sessionClosed(NextFilter nextFilter, IoSession session) {
            log.warn("register session closed");
            ACTIVE_SESSION.set(null);
            nextFilter.sessionClosed(session);
        }
    }

}
